/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server.quorum;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.zookeeper.jmx.MBeanRegistry;
import org.apache.zookeeper.server.ZooKeeperThread;
import org.apache.zookeeper.server.quorum.QuorumCnxManager.Message;
import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Implementation of leader election using TCP. It uses an object of the class
 * QuorumCnxManager to manage connections. Otherwise, the algorithm is
 * push-based as with the other UDP implementations.
 *
 * There are a few parameters that can be tuned to change its behavior. First,
 * finalizeWait determines the amount of time to wait until deciding upon a
 * leader. This is part of the leader election algorithm.
 */

public class FastLeaderElection implements Election {
    private static final Logger LOG = LoggerFactory.getLogger(FastLeaderElection.class);

    /**
     * Determine how much time a process has to wait once it believes that it
     * has reached the end of leader election.
     */
    final static int finalizeWait = 200;

    /**
     * Upper bound on the amount of time between two consecutive notification
     * checks. This impacts the amount of time to get the system up again after
     * long partitions. Currently 60 seconds.
     */

    final static int maxNotificationInterval = 60000;

    /**
     * Connection manager. Fast leader election uses TCP for communication
     * between peers, and QuorumCnxManager manages such connections.
     */

    QuorumCnxManager manager;

    /**
     * Notifications are messages that let other peers know that a given peer
     * has changed its vote, either because it has joined leader election or
     * because it learned of another peer with higher zxid or same zxid and
     * higher server id
     */

    static public class Notification {
        /*
         * Format version, introduced in 3.4.6
         */

        public final static int CURRENTVERSION = 0x1;
        int version;

        /**
         * 提议的leader
         */
        long leader;

        /*
         * 事务id，事务请求的唯一标记，由leader服务器负责进行分配。 高32位是peerEpoch，低32位是请求的计数，从0开始。
         */
        long zxid;

        /*
         * 每次leader选举，electionEpoch就会自增1，统计投票信息时，首先保证electionEpoch相同。
         */
        long electionEpoch;

        /*
         * current state of sender
         */
        QuorumPeer.ServerState state;

        /*
         * 服务器id
         */
        long sid;

        /*
         * 每次leader选举完成之后，都会选举出一个新的peerEpoch，用来标记事务请求所属的轮次
         */
        long peerEpoch;

        @Override
        public String toString() {
            return new String(Long.toHexString(version) + " (message format version), " + leader + " (n.leader), 0x"
                    + Long.toHexString(zxid) + " (n.zxid), 0x" + Long.toHexString(electionEpoch) + " (n.round), "
                    + state + " (n.state), " + sid + " (n.sid), 0x" + Long.toHexString(peerEpoch) + " (n.peerEpoch) ");
        }
    }

    static ByteBuffer buildMsg(int state, long leader, long zxid, long electionEpoch, long epoch) {
        byte requestBytes[] = new byte[40];
        ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);

        /*
         * Building notification packet to send
         */

        requestBuffer.clear();
        requestBuffer.putInt(state);
        requestBuffer.putLong(leader);
        requestBuffer.putLong(zxid);
        requestBuffer.putLong(electionEpoch);
        requestBuffer.putLong(epoch);
        requestBuffer.putInt(Notification.CURRENTVERSION);

        return requestBuffer;
    }

    /**
     * Messages that a peer wants to send to other peers. These messages can be
     * both Notifications and Acks of reception of notification.
     */
    static public class ToSend {
        static enum mType {
            crequest, challenge, notification, ack
        }

        ToSend(mType type, long leader, long zxid, long electionEpoch, ServerState state, long sid, long peerEpoch) {

            this.leader = leader;
            this.zxid = zxid;
            this.electionEpoch = electionEpoch;
            this.state = state;
            this.sid = sid;
            this.peerEpoch = peerEpoch;
        }

        /*
         * Proposed leader in the case of notification
         */
        long leader;

        /*
         * id contains the tag for acks, and zxid for notifications
         */
        long zxid;

        /*
         * Epoch
         */
        long electionEpoch;

        /*
         * Current state;
         */
        QuorumPeer.ServerState state;

        /*
         * Address of recipient
         */
        long sid;

        /*
         * Leader epoch
         */
        long peerEpoch;
    }

    LinkedBlockingQueue<ToSend> sendqueue;
    LinkedBlockingQueue<Notification> recvqueue;

    /**
     * Multi-threaded implementation of message handler. Messenger implements
     * two sub-classes: WorkReceiver and WorkSender. The functionality of each
     * is obvious from the name. Each of these spawns a new thread.
     */

    protected class Messenger {

        /**
         * Receives messages from instance of QuorumCnxManager on method run(),
         * and processes such messages.
         */

        class WorkerReceiver extends ZooKeeperThread {
            volatile boolean stop;
            QuorumCnxManager manager;

            WorkerReceiver(QuorumCnxManager manager) {
                super("WorkerReceiver");
                this.stop = false;
                this.manager = manager;
            }

            public void run() {

                Message response;
                while (!stop) {
                    // Sleeps on receive
                    try {
                        response = manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
                        System.err.println("WorkerReceiver-------sid="+(response==null ? 0 : response.sid));
                        if (response == null)
                            continue;

                        /*
                         * If it is from an observer, respond right away. Note
                         * that the following predicate assumes that if a server
                         * is not a follower, then it must be an observer. If we
                         * ever have any other type of learner in the future,
                         * we'll have to change the way we check for observers.
                         */
                        if (!self.getVotingView().containsKey(response.sid)) {
                            Vote current = self.getCurrentVote();
                            ToSend notmsg = new ToSend(ToSend.mType.notification, current.getId(), current.getZxid(),
                                    logicalclock, self.getPeerState(), response.sid, current.getPeerEpoch());

                            sendqueue.offer(notmsg);
                        } else {
                            // Receive new message
                            if (LOG.isDebugEnabled()) {
                                LOG.debug("Receive new notification message. My id = " + self.getId());
                            }

                            /*
                             * We check for 28 bytes for backward compatibility
                             */
                            if (response.buffer.capacity() < 28) {
                                LOG.error("Got a short response: " + response.buffer.capacity());
                                continue;
                            }
                            boolean backCompatibility = (response.buffer.capacity() == 28);
                            response.buffer.clear();

                            // Instantiate Notification and set its attributes
                            Notification n = new Notification();

                            // State of peer that sent this message
                            QuorumPeer.ServerState ackstate = QuorumPeer.ServerState.LOOKING;
                            switch (response.buffer.getInt()) {
                            case 0:
                                ackstate = QuorumPeer.ServerState.LOOKING;
                                break;
                            case 1:
                                ackstate = QuorumPeer.ServerState.FOLLOWING;
                                break;
                            case 2:
                                ackstate = QuorumPeer.ServerState.LEADING;
                                break;
                            case 3:
                                ackstate = QuorumPeer.ServerState.OBSERVING;
                                break;
                            default:
                                continue;
                            }

                            n.leader = response.buffer.getLong();
                            n.zxid = response.buffer.getLong();
                            n.electionEpoch = response.buffer.getLong();
                            n.state = ackstate;
                            n.sid = response.sid;
                            if (!backCompatibility) {
                                n.peerEpoch = response.buffer.getLong();
                            } else {
                                if (LOG.isInfoEnabled()) {
                                    LOG.info("Backward compatibility mode, server id=" + n.sid);
                                }
                                n.peerEpoch = ZxidUtils.getEpochFromZxid(n.zxid);
                            }

                            /*
                             * Version added in 3.4.6
                             */

                            n.version = (response.buffer.remaining() >= 4) ? response.buffer.getInt() : 0x0;

                            /*
                             * Print notification info
                             */
                            if (LOG.isInfoEnabled()) {
                                printNotification(n);
                            }

                            /*
                             * If this server is looking, then send proposed
                             * leader
                             */

                            if (self.getPeerState() == QuorumPeer.ServerState.LOOKING) {
                                recvqueue.offer(n);

                                /*
                                 * Send a notification back if the peer that
                                 * sent this message is also looking and its
                                 * logical clock is lagging behind.
                                 */
                                if ((ackstate == QuorumPeer.ServerState.LOOKING) && (n.electionEpoch < logicalclock)) {
                                    Vote v = getVote();
                                    ToSend notmsg = new ToSend(ToSend.mType.notification, v.getId(), v.getZxid(),
                                            logicalclock, self.getPeerState(), response.sid, v.getPeerEpoch());
                                    sendqueue.offer(notmsg);
                                }
                            } else {
                                /*
                                 * If this server is not looking, but the one
                                 * that sent the ack is looking, then send back
                                 * what it believes to be the leader.
                                 */
                                Vote current = self.getCurrentVote();
                                if (ackstate == QuorumPeer.ServerState.LOOKING) {
                                    if (LOG.isDebugEnabled()) {
                                        LOG.debug("Sending new notification. My id =  " + self.getId() + " recipient="
                                                + response.sid + " zxid=0x" + Long.toHexString(current.getZxid())
                                                + " leader=" + current.getId());
                                    }

                                    ToSend notmsg;
                                    if (n.version > 0x0) {
                                        notmsg = new ToSend(ToSend.mType.notification, current.getId(),
                                                current.getZxid(), current.getElectionEpoch(), self.getPeerState(),
                                                response.sid, current.getPeerEpoch());

                                    } else {
                                        Vote bcVote = self.getBCVote();
                                        notmsg = new ToSend(ToSend.mType.notification, bcVote.getId(), bcVote.getZxid(),
                                                bcVote.getElectionEpoch(), self.getPeerState(), response.sid,
                                                bcVote.getPeerEpoch());
                                    }
                                    sendqueue.offer(notmsg);
                                }
                            }
                        }
                    } catch (InterruptedException e) {
                        System.out.println("Interrupted Exception while waiting for new message" + e.toString());
                    }
                }
                LOG.info("WorkerReceiver is down");
            }
        }

        /**
         * This worker simply dequeues a message to send and and queues it on
         * the manager's queue.
         */

        class WorkerSender extends ZooKeeperThread {
            volatile boolean stop;
            QuorumCnxManager manager;

            WorkerSender(QuorumCnxManager manager) {
                super("WorkerSender");
                this.stop = false;
                this.manager = manager;
            }

            public void run() {
                while (!stop) {
                    try {
                        ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS);
                        if (m == null)
                            continue;

                        process(m);
                    } catch (InterruptedException e) {
                        break;
                    }
                }
                LOG.info("WorkerSender is down");
            }

            /**
             * Called by run() once there is a new message to send.
             *
             * @param m
             *            message to send
             */
            void process(ToSend m) {
                ByteBuffer requestBuffer = buildMsg(m.state.ordinal(), m.leader, m.zxid, m.electionEpoch, m.peerEpoch);
                manager.toSend(m.sid, requestBuffer);
            }
        }

        /**
         * Test if both send and receive queues are empty.
         */
        public boolean queueEmpty() {
            return (sendqueue.isEmpty() || recvqueue.isEmpty());
        }

        WorkerSender ws;
        WorkerReceiver wr;

        /**
         * Constructor of class Messenger.
         *
         * @param manager
         *            Connection manager
         */
        Messenger(QuorumCnxManager manager) {

            this.ws = new WorkerSender(manager);

            Thread t = new Thread(this.ws, "WorkerSender[myid=" + self.getId() + "]");
            t.setDaemon(true);
            t.start();

            this.wr = new WorkerReceiver(manager);

            t = new Thread(this.wr, "WorkerReceiver[myid=" + self.getId() + "]");
            t.setDaemon(true);
            t.start();
        }

        /**
         * Stops instances of WorkerSender and WorkerReceiver
         */
        void halt() {
            this.ws.stop = true;
            this.wr.stop = true;
        }

    }

    QuorumPeer self;
    Messenger messenger;
    volatile long logicalclock; /* Election instance */
    long proposedLeader;
    long proposedZxid;
    long proposedEpoch;

    /**
     * Returns the current vlue of the logical clock counter
     */
    public long getLogicalClock() {
        return logicalclock;
    }

    /**
     * Constructor of FastLeaderElection. It takes two parameters, one is the
     * QuorumPeer object that instantiated this object, and the other is the
     * connection manager. Such an object should be created only once by each
     * peer during an instance of the ZooKeeper service.
     *
     * @param self
     *            QuorumPeer that created this object
     * @param manager
     *            Connection manager
     */
    public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager) {
        this.stop = false;
        this.manager = manager;
        starter(self, manager);
    }

    /**
     * This method is invoked by the constructor. Because it is a part of the
     * starting procedure of the object that must be on any constructor of this
     * class, it is probably best to keep as a separate method. As we have a
     * single constructor currently, it is not strictly necessary to have it
     * separate.
     *
     * @param self
     *            QuorumPeer that created this object
     * @param manager
     *            Connection manager
     */
    private void starter(QuorumPeer self, QuorumCnxManager manager) {
        this.self = self;
        proposedLeader = -1;
        proposedZxid = -1;

        sendqueue = new LinkedBlockingQueue<ToSend>();
        recvqueue = new LinkedBlockingQueue<Notification>();
        this.messenger = new Messenger(manager);
    }

    private void leaveInstance(Vote v) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("About to leave FLE instance: leader=" + v.getId() + ", zxid=0x" + Long.toHexString(v.getZxid())
                    + ", my id=" + self.getId() + ", my state=" + self.getPeerState());
        }
        recvqueue.clear();
    }

    public QuorumCnxManager getCnxManager() {
        return manager;
    }

    volatile boolean stop;

    public void shutdown() {
        stop = true;
        LOG.debug("Shutting down connection manager");
        manager.halt();
        LOG.debug("Shutting down messenger");
        messenger.halt();
        LOG.debug("FLE is down");
    }

    /**
     * Send notifications to all peers upon a change in our vote
     */
    private void sendNotifications() {
        for (QuorumServer server : self.getVotingView().values()) {
            long sid = server.id;

            ToSend notmsg = new ToSend(ToSend.mType.notification, proposedLeader, proposedZxid, logicalclock,
                    QuorumPeer.ServerState.LOOKING, sid, proposedEpoch);
            if (LOG.isDebugEnabled()) {
                LOG.debug("Sending Notification: " + proposedLeader + " (n.leader), 0x" + Long.toHexString(proposedZxid)
                        + " (n.zxid), 0x" + Long.toHexString(logicalclock) + " (n.round), " + sid + " (recipient), "
                        + self.getId() + " (myid), 0x" + Long.toHexString(proposedEpoch) + " (n.peerEpoch)");
            }
            sendqueue.offer(notmsg);
        }
    }

    private void printNotification(Notification n) {
        LOG.info("Notification: " + n.toString() + self.getPeerState() + " (my state)");
    }

    /**
     * 如果new票信息大于当前的票信息，那么返回true。以下三种case，只要满足一种就返回true：
     * <li>newEpoch > curEpoch
     * <li>newEpoch == curEpoch && newZxid > curZxid
     * <li>newEpoch == curEpoch && newZxid = curZxid && newId > curId
     * 
     * @param newId
     * @param newZxid
     * @param newEpoch
     * @param curId
     * @param curZxid
     * @param curEpoch
     * @return
     */
    protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid,
            long curEpoch) {
        LOG.debug("id: " + newId + ", proposed id: " + curId + ", zxid: 0x" + Long.toHexString(newZxid)
                + ", proposed zxid: 0x" + Long.toHexString(curZxid));
        if (self.getQuorumVerifier().getWeight(newId) == 0) {
            return false;
        }

        /*
         * We return true if one of the following three cases hold: 1- New epoch
         * is higher 2- New epoch is the same as current epoch, but new zxid is
         * higher 3- New epoch is the same as current epoch, new zxid is the
         * same as current zxid, but server id is higher.
         */
        return ((newEpoch > curEpoch)
                || ((newEpoch == curEpoch) && ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId)))));
    }

    /**
     * 是否终止投票，如果{@link vote}对应的服务器获得了超过半数的票，那么返回true
     * 
     * @param votes
     * @param vote
     * @return
     */
    protected boolean termPredicate(HashMap<Long, Vote> votes, Vote vote) {

        HashSet<Long> set = new HashSet<Long>();

        /*
         * First make the views consistent. Sometimes peers will have different
         * zxids for a server depending on timing.
         */
        for (Map.Entry<Long, Vote> entry : votes.entrySet()) {
            if (vote.equals(entry.getValue())) {
                set.add(entry.getKey());
            }
        }

        return self.getQuorumVerifier().containsQuorum(set);
    }

    /**
     * leader确认检查
     * <li>如果自己不是leader，那么一定要收到过Leader的信息，并且leader宣称自己的状态是ServerState.LEADING；
     * <li>如果自己是leader，那么当前logicalclock一定要等于投票信息中的electionEpoch
     * 
     * @param outofelection
     * @param leader
     * @param electionEpoch
     */
    protected boolean checkLeader(HashMap<Long, Vote> outofelection, long leader, long electionEpoch) {

        boolean predicate = true;

        /*
         * If everyone else thinks I'm the leader, I must be the leader. The
         * other two checks are just for the case in which I'm not the leader.
         * If I'm not the leader and I haven't received a message from leader
         * stating that it is leading, then predicate is false.
         */

        if (leader != self.getId()) {
            if (outofelection.get(leader) == null)
                // 表示还未从leader中收到任何信息，所以选举结束未结束。
                predicate = false;
            else if (outofelection.get(leader).getState() != ServerState.LEADING)
                // 指推举的那个leader还没有将自己的状态改成leader，所以选举依然未完成。
                predicate = false;
        } else if (logicalclock != electionEpoch) {
            // 自己是leader时，这两个值应该是相同的，如果不同说明有异常，不能结束选举。
            predicate = false;
        }

        return predicate;
    }

    /**
     * 是否能宣称选举结束
     * <li>{@code n}获得了半数以上的票
     * <li>通过了leader确认的检查
     * 
     * @param recv
     *            map of received votes
     * @param ooe
     *            map containing out of election votes (LEADING or FOLLOWING)
     * @param n
     *            Notification
     * @return
     */
    protected boolean ooePredicate(HashMap<Long, Vote> recv, HashMap<Long, Vote> ooe, Notification n) {

        return (termPredicate(recv, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state))
                && checkLeader(ooe, n.leader, n.electionEpoch));

    }

    synchronized void updateProposal(long leader, long zxid, long epoch) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Updating proposal: " + leader + " (newleader), 0x" + Long.toHexString(zxid) + " (newzxid), "
                    + proposedLeader + " (oldleader), 0x" + Long.toHexString(proposedZxid) + " (oldzxid)");
        }
        proposedLeader = leader;
        proposedZxid = zxid;
        proposedEpoch = epoch;
    }

    synchronized Vote getVote() {
        return new Vote(proposedLeader, proposedZxid, proposedEpoch);
    }

    /**
     * A learning state can be either FOLLOWING or OBSERVING. This method simply
     * decides which one depending on the role of the server.
     *
     * @return ServerState
     */
    private ServerState learningState() {
        if (self.getLearnerType() == LearnerType.PARTICIPANT) {
            LOG.debug("I'm a participant: " + self.getId());
            return ServerState.FOLLOWING;
        } else {
            LOG.debug("I'm an observer: " + self.getId());
            return ServerState.OBSERVING;
        }
    }

    /**
     * Returns the initial vote value of server identifier.
     *
     * @return long
     */
    private long getInitId() {
        if (self.getLearnerType() == LearnerType.PARTICIPANT)
            return self.getId();
        else
            return Long.MIN_VALUE;
    }

    /**
     * Returns initial last logged zxid.
     *
     * @return long
     */
    private long getInitLastLoggedZxid() {
        if (self.getLearnerType() == LearnerType.PARTICIPANT)
            return self.getLastLoggedZxid();
        else
            return Long.MIN_VALUE;
    }

    /**
     * Returns the initial vote value of the peer epoch.
     *
     * @return long
     */
    private long getPeerEpoch() {
        if (self.getLearnerType() == LearnerType.PARTICIPANT)
            try {
                return self.getCurrentEpoch();
            } catch (IOException e) {
                RuntimeException re = new RuntimeException(e.getMessage());
                re.setStackTrace(e.getStackTrace());
                throw re;
            }
        else
            return Long.MIN_VALUE;
    }

    /**
     * Starts a new round of leader election. Whenever our QuorumPeer changes
     * its state to LOOKING, this method is invoked, and it sends notifications
     * to all other peers.
     */
    public Vote lookForLeader() throws InterruptedException {
        try {
            self.jmxLeaderElectionBean = new LeaderElectionBean();
            MBeanRegistry.getInstance().register(self.jmxLeaderElectionBean, self.jmxLocalPeerBean);
        } catch (Exception e) {
            LOG.warn("Failed to register with JMX", e);
            self.jmxLeaderElectionBean = null;
        }
        if (self.start_fle == 0) {
            self.start_fle = System.currentTimeMillis();
        }
        try {
            // 投票信息列表
            HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();

            HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();

            int notTimeout = finalizeWait;

            synchronized (this) {
                // 更新选举周期
                logicalclock++;
                // 更新提议信息，提议自己作为leader，
                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
            }

            LOG.info("New election. My id =  " + self.getId() + ", proposed zxid=0x" + Long.toHexString(proposedZxid));
            // 发送投票信息
            sendNotifications();

            /*
             * 循环直到选举出leader
             */
            while ((self.getPeerState() == ServerState.LOOKING) && (!stop)) {
                // 从recvqueue移除第一个Notification（投票信息），最多等待notTimeout ms
                Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);

                // 如果当前没有收到足够多的Notification，不足以产生leader，那么继续发送投票信息，否则继续处理收到的Notification
                if (n == null) {
                    // 当前没有收到回复信息
                    if (manager.haveDelivered()) {
                        // 如果消息全部投递出去了，那么在发送一次投票信息
                        sendNotifications();
                    } else {
                        // 如果消息没有投递出去，那么尝试重连
                        manager.connectAll();
                    }

                    int tmpTimeOut = notTimeout * 2;
                    notTimeout = (tmpTimeOut < maxNotificationInterval ? tmpTimeOut : maxNotificationInterval);
                    LOG.info("Notification time out: " + notTimeout);
                } else if (self.getVotingView().containsKey(n.sid)) {
                    /*
                     * Only proceed if the vote comes from a replica in the
                     * voting view.
                     */
                    switch (n.state) {
                    case LOOKING:
                        // If notification > current, replace and send messages
                        // out
                        if (n.electionEpoch > logicalclock) {
                            logicalclock = n.electionEpoch;
                            // 清空收到的投票信息列表
                            recvset.clear();
                            // 比较notification，current，大的最为下次推荐的leader，然后发送投票信息
                            if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, getInitId(), getInitLastLoggedZxid(),
                                    getPeerEpoch())) {
                                updateProposal(n.leader, n.zxid, n.peerEpoch);
                            } else {
                                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
                            }
                            sendNotifications();
                        } else if (n.electionEpoch < logicalclock) {
                            // 如果notification#electionEpoch小,认为是无效投票，直接忽略
                            if (LOG.isDebugEnabled()) {
                                LOG.debug(
                                        "Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x"
                                                + Long.toHexString(n.electionEpoch) + ", logicalclock=0x"
                                                + Long.toHexString(logicalclock));
                            }
                            break;
                        } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid,
                                proposedEpoch)) {
                            // 比较notification，current，大的最为下次推荐的leader，然后发送投票信息
                            updateProposal(n.leader, n.zxid, n.peerEpoch);
                            sendNotifications();
                        }

                        if (LOG.isDebugEnabled()) {
                            LOG.debug("Adding vote: from=" + n.sid + ", proposed leader=" + n.leader
                                    + ", proposed zxid=0x" + Long.toHexString(n.zxid) + ", proposed election epoch=0x"
                                    + Long.toHexString(n.electionEpoch));
                        }

                        // 将收到的投票加入到投票列表中
                        recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));

                        // 判断self提议的服务器是否已经获得超过半数的票
                        if (termPredicate(recvset,
                                new Vote(proposedLeader, proposedZxid, logicalclock, proposedEpoch))) {

                            // 循环直到当recvqueue中没有Notification，或有出现一个Notification的投票信息比推举的leader更大时，方跳出循环
                            while ((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null) {
                                if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid,
                                        proposedEpoch)) {
                                    recvqueue.put(n);
                                    break;
                                }
                            }

                            if (n == null) {
                                // 进入这里说明recvqueue中没有比推举leader更大的票，可以将推举的服务器作为leader了。
                                self.setPeerState(
                                        (proposedLeader == self.getId()) ? ServerState.LEADING : learningState());

                                Vote endVote = new Vote(proposedLeader, proposedZxid, logicalclock, proposedEpoch);
                                leaveInstance(endVote);
                                return endVote;
                            }
                        }
                        break;
                    case OBSERVING:
                        LOG.debug("Notification from observer: " + n.sid);
                        break;
                    case FOLLOWING:
                    case LEADING:
                        /**
                         * if内的代码、if外的代码比较类似，唯一的差别是ooePredicate方法的第一个参数，
                         * 考虑集群中只有leader存活，其他所有服务器都处于LOOKING状态，就能理解为什么需要这两段逻辑了。
                         */
                        if (n.electionEpoch == logicalclock) {
                            // 如果Notification的electionEpoch和当前的electionEpoch相同，那么说明在同一轮的选举中
                            recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));

                            // 判定选举是否结束
                            if (ooePredicate(recvset, outofelection, n)) {
                                // 选举结束，设置状态
                                self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());

                                Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
                                leaveInstance(endVote);
                                return endVote;
                            }
                        }

                        outofelection.put(n.sid,
                                new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
                        // 判定选举是否结束
                        if (ooePredicate(outofelection, outofelection, n)) {
                            synchronized (this) {
                                logicalclock = n.electionEpoch;
                                self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());
                            }
                            Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
                            leaveInstance(endVote);
                            return endVote;
                        }
                        break;
                    default:
                        LOG.warn("Notification state unrecognized: {} (n.state), {} (n.sid)", n.state, n.sid);
                        break;
                    }
                } else {
                    LOG.warn("Ignoring notification from non-cluster member " + n.sid);
                }
            }
            return null;
        } finally {
            try {
                if (self.jmxLeaderElectionBean != null) {
                    MBeanRegistry.getInstance().unregister(self.jmxLeaderElectionBean);
                }
            } catch (Exception e) {
                LOG.warn("Failed to unregister with JMX", e);
            }
            self.jmxLeaderElectionBean = null;
            LOG.debug("Number of connection processing threads: {}", manager.getConnectionThreadCount());
        }
    }
}
